package com.easou.ps.master.rpc;

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.easou.ps.common.servicehandle.Call;
import com.easou.ps.common.servicehandle.ConnectionGuard;
import com.easou.ps.common.servicehandle.ServiceAddress;
import com.easou.ps.master.kit.ConfigFactory;
import com.easou.ps.thrift.ks.KSRPC.search_args;
import com.easou.ps.thrift.ks.KSRPC.search_result;
import com.easou.ps.thrift.ks.KnowledgeSearchRequest;
import com.easou.ps.thrift.ks.KnowledgeSearchResponse;

public class KSRPCProxy {
	
	private static final Logger LOG = Logger.getLogger("search");
	private static final int SEARCH_TIMEOUT = ConfigFactory.getSysConfig().getInt("rpc.search.timeout", 8000);

	private final ServiceAddress address;
	
	AtomicInteger idGenerator = new AtomicInteger(0);
	
	public KSRPCProxy(ServiceAddress address) {
		this.address = address;
	}
	
	public KnowledgeSearchResponse search(KnowledgeSearchRequest query) {
		int seqid = idGenerator.getAndIncrement();
		String func = "search";
		
		search_args args = new search_args();
		args.setReq(query);
		
		search_result result = new search_result();

		Call call = new Call(seqid, func, args, result);
		
		KnowledgeSearchResponse ret = null;
		ConnectionGuard conn = address.getConnection(seqid);
		if (conn == null) {
			LOG.error("Worker not running:" + address);
			return ret;
		}
		if (!conn.sendCall(call)) {
			LOG.error("send request to worker fail, worker[" + address + "], req:" + query);
		} else {
			call.waitComplete(SEARCH_TIMEOUT);
			if (call.isFail()) {
//				address.fail(conn);
				LOG.error("error occured when waiting search result, worker[" + address + "], req:" + query);
			} else if (call.isTimeout()) {
//				address.timeout(conn);
				LOG.error("timeout when waiting search result, worker[" + address + "], req:" + query);
			} else {
				ret = ((search_result)call.getResult()).success;
			}
		}
		return ret;
	}
	
	@Override
	public String toString(){
		return this.address.toString();
	}
}
